package cn.doitedu.hbase.baseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @date: 2019/7/9
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description:  hbase客户端api的数据定义操作
 */
public class DDLDemos {
    Connection conn = null;

    @Before
    public void init() throws Exception {

        //Configuration conf = new Configuration();  它会自动加载classpath中的core-site.xml core-default.xml hdfs-site.xml hdfs-default.xml yarn-default.xml yarn-site.xml mapred-default.xml mapred-site.xml
        Configuration conf = HBaseConfiguration.create(); // 不光会加载hadoop的配置，还会加载  hbase-site.xml

        // 客户端连接不需要指定具体的master或者regionserver地址，只要指定zookeeper地址就行
        conf.set("hbase.zookeeper.quorum","spark01:2181,spark02:2181,spark03:2181");

        conn = ConnectionFactory.createConnection(conf);

    }


    /**
     * 创建名称空间
     * @throws Exception
     */
    @Test
    public void testCreateNameSpace() throws Exception {


        // 获取表定义管理器
        Admin admin = conn.getAdmin();

        NamespaceDescriptor yourspace = NamespaceDescriptor.create("yourspace").build();
        admin.createNamespace(yourspace);

        admin.close();
        conn.close();
    }


    /**
     * 创建表
     * @throws IOException
     */
    @Test
    public void testCreateTable() throws IOException {

        Admin admin = conn.getAdmin();

        // 构建一个表定义描述对象创建器
        TableDescriptorBuilder tbBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("yourspace", "t5"));

        // 获取一个列族描述对象构建器
        ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder("f".getBytes());
        // 为列族定义设置各种参数
        cfBuilder.setBloomFilterType(BloomType.ROWCOL);
        cfBuilder.setTimeToLive(5000);
        cfBuilder.setMaxVersions(3);
        // 用列族构建器构建一个列族定义描述对象
        ColumnFamilyDescriptor cf = cfBuilder.build();

        // 用表构建器设置列族，并构建表描述对象
        TableDescriptor td = tbBuilder.setColumnFamily(cf).build();

        // 用admin管理器创建表
        admin.createTable(td);

        admin.close();
        conn.close();
    }


    /**
     * 创建预分区表
     * @throws IOException
     */
    @Test
    public void testCreateTableSplit() throws IOException {

        Admin admin = conn.getAdmin();

        TableDescriptorBuilder t6Builder = TableDescriptorBuilder.newBuilder(TableName.valueOf("t6"));

        ColumnFamilyDescriptor f  = ColumnFamilyDescriptorBuilder.newBuilder("f".getBytes()).build();

        TableDescriptor t6 = t6Builder.setColumnFamily(f).build();


        byte[][] splitKeys = {"a".getBytes(),"f".getBytes(),"k".getBytes(),"q".getBytes(),"v".getBytes()};


        // 在这里指定预分区的分界点keys
        admin.createTable(t6,splitKeys);

        admin.close();
        conn.close();


    }


    /**
     * 表定义修改
     * @throws Exception
     */
    @Test
    public void modifyTableDescription() throws Exception {

        Admin admin = conn.getAdmin();

        /**
         * 示例1： 修改之前的列定义
         */
        TableDescriptor td = admin.getDescriptor(TableName.valueOf("yourspace".getBytes(), "t5".getBytes()));
        // 从表定义中取出列族f的定义
        ColumnFamilyDescriptor cfDesc = td.getColumnFamily("f".getBytes());
        // 用列族定义构建器对原来的列族f定义进行修改
        ColumnFamilyDescriptor cfDesc2 = ColumnFamilyDescriptorBuilder.newBuilder(cfDesc).setTimeToLive(Integer.MAX_VALUE).build();

        // 修改指定表的列族定义
        admin.modifyColumnFamily(TableName.valueOf("yourspace".getBytes(), "t5".getBytes()),cfDesc2);


        /**
         * 示例2：增加一个新的列定义
         */
        // 构建一个新的列族定义
        ColumnFamilyDescriptor cf_b = ColumnFamilyDescriptorBuilder.newBuilder("b".getBytes()).build();
        td = admin.getDescriptor(TableName.valueOf("yourspace".getBytes(), "t5".getBytes()));
        // 对原来的表定义，设置一个新的b列族
        TableDescriptorBuilder tBuilder = TableDescriptorBuilder.newBuilder(td);
        tBuilder.setColumnFamily(cf_b);
        TableDescriptor td2 = tBuilder.build();

        // 通过客户端发送表定义修改命令
        admin.modifyTable(td2);

        admin.close();
        conn.close();
    }


    /**
     * 删除表的列族定义、表定义、名称空间定义
     * @throws Exception
     */
    @Test
    public void delete() throws Exception {


        Admin admin = conn.getAdmin();

        // 这个是表定义中的删除列族，它会把这个列族的定义直接删掉（当然包含数据）
        //admin.deleteColumnFamily(TableName.valueOf("t1"),"extra_info".getBytes());


        // 禁用一个表
        admin.disableTable(TableName.valueOf("yourspace".getBytes(),"t5".getBytes()));


        // 删除整个表：包含表定义和数据
        admin.deleteTable(TableName.valueOf("yourspace".getBytes(),"t5".getBytes()));

        // 要删除名称空间，必须先把其中的表全部删除
        admin.deleteNamespace("yourspace");


        admin.close();
        conn.close();


    }


    /**
     *
     * 列出所有名称空间/表
     * @throws Exception
     */
    @Test
    public void listTables() throws Exception {

        Admin admin = conn.getAdmin();


        /**
         * 列名称空间
         */
        NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
        for (NamespaceDescriptor namespaceDescriptor : namespaceDescriptors) {
            System.out.println(namespaceDescriptor.getName());
        }


        /**
         * 列表
         */
        TableName[] tableNames = admin.listTableNames();
        for (TableName tableName : tableNames) {
            System.out.println(tableName.getNameAsString());
        }
        admin.close();
        conn.close();
    }
}
